package com.atguigu;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class Hbase_DML {
    public static Connection connection = Hbase_Single_Connect.connection;


    //写数据
    public static void putCell(String namespace, String tableName, String rowKey, String columFamily, String colum, String value) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.调用put方法写数据
        Put put = new Put(Bytes.toBytes(rowKey));

        put.addColumn(Bytes.toBytes(columFamily), Bytes.toBytes(colum), Bytes.toBytes(value));


        table.put(put);

        //关闭链接
        table.close();

    }


    //获取数据
    public static void getCell(String namespace, String tableName, String rowKey, String columFamily, String colum) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));


        //2.调用get方法获取数据
        Get get = new Get(Bytes.toBytes(rowKey));

        //获取指定列族下的数据
        get.addFamily(Bytes.toBytes(columFamily));
        //获取指定列下的数据
//        get.addColumn(Bytes.toBytes(columFamily), Bytes.toBytes(colum));

        Result result = table.get(get);

        for (Cell cell : result.rawCells()) {
            System.out.println(
                    //获取rowkey
                    Bytes.toString(CellUtil.cloneRow(cell)) + "-" +
                            //获取列族
                            Bytes.toString(CellUtil.cloneFamily(cell)) + "-" +
                            //获取列
                            Bytes.toString(CellUtil.cloneQualifier(cell)) + "-" +
                            //获取数据
                            Bytes.toString(CellUtil.cloneValue(cell)));
        }

        //打印结果
//        System.out.println(Bytes.toString(result.value()));

        //关闭链接
        table.close();
    }

    //删除数据
    public static void deleteCell(String namespace, String tableName, String rowKey, String columFamily, String colum) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.调用delete删除
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        //指定删除列的最新版本的数据
//        delete.addColumn(Bytes.toBytes(columFamily), Bytes.toBytes(colum));
        //指定删除列的所有版本的数据 等同于 shell命令中的deleteall
        delete.addColumns(Bytes.toBytes(columFamily), Bytes.toBytes(colum));

        table.delete(delete);


        //关闭链接
        table.close();


    }


    //扫描数据


    //关闭链接
    public static void scanValues(String namespace, String tableName, String startRowKey, String stopRowKey) throws IOException {

        //1.获取表的链接
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));


        //2.调用scan方法扫描数据
        Scan scan = new Scan();

        scan.setRaw(true);
        scan.readAllVersions();
//        scan.readVersions(3);

        //指定起始rowkey
        scan.withStartRow(Bytes.toBytes(startRowKey));
        //指定末尾rowkey
        scan.withStopRow(Bytes.toBytes(stopRowKey),true);

        ResultScanner scanner = table.getScanner(scan);

        //遍历获取Result
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(
                        //获取rowkey
                        Bytes.toString(CellUtil.cloneRow(cell)) + "-" +
                                //获取列族
                                Bytes.toString(CellUtil.cloneFamily(cell)) + "-" +
                                //获取列
                                Bytes.toString(CellUtil.cloneQualifier(cell)) + "-" +
                                //获取数据
                                Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        table.close();


    }


    public static void main(String[] args) throws IOException {
//        putCell("bigdata", "student", "1001", "info", "name", "zs");
//        putCell("bigdata", "student", "1001", "info", "age", "18");
//        putCell("bigdata", "student", "1001", "info", "age", "19");
//        putCell("bigdata", "student", "1001", "info", "age", "20");
//        putCell("bigdata", "student", "1001", "info", "age", "21");
//        putCell("bigdata", "student", "1001", "info", "age", "22");
//        putCell("bigdata", "student", "1001", "info", "age", "23");
//        putCell("bigdata", "student", "1001", "info", "age", "24");
//        putCell("bigdata", "student", "1001", "info", "age", "25");
//        putCell("bigdata", "student", "1002", "info", "name", "ls");
//        putCell("bigdata", "student", "1002", "info", "age", "19");
//        putCell("bigdata", "student", "1003", "info", "name", "ww");
//        putCell("bigdata", "student", "1003", "info", "age", "20");
//        putCell("bigdata", "student", "1004", "info", "name", "xiaoming");
//        putCell("bigdata", "student", "1004", "info", "age", "23");

//        deleteCell("bigdata", "student", "1001", "info", "age");
//        getCell("bigdata", "student", "1001", "info", "name");

        scanValues("bigdata", "student", "1001", "1003");
        //关闭Hbase链接
        Hbase_Single_Connect.closeConnect();
    }
}
